package DataStreamApi.Sink输出;


import domain.WaterSensor;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;

import java.util.Arrays;
import java.util.List;

/**
 * 往KAFKA中写入数据
 */
public class Flink02_KAFKA {
    public static void main(String[] args) throws Exception {

        /**
         * 创建执行环境
         * IDEA运行的时候，
         * 需要导入依赖
         */
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        /**
         * 读取数据，此处为从文件中读取数据
         */

        DataStreamSource<String> stringDataStreamSource = executionEnvironment.socketTextStream("192.168.159.104", 7777);
        //nc -l -p 7777


        KafkaSink.<String>builder()
                //指定kafka的地址和端口
                .setBootstrapServers("114.251.235.19:9092")
                //指定序列化器
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("topic_test")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                //指定Kafka的一致性级别：精准一次，至少一次
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                //如果是精准一次，必须设置事务前缀。
                .build();


        executionEnvironment.execute();


    }
}
